package com.gitee.transformation

import com.gitee.source.practice
import com.gitee.source.practice.CustomerSourceMySQL
import org.apache.flink.streaming.api.scala.{DataStream, SplitStream, StreamExecutionEnvironment, _}

import scala.collection.mutable.ArrayBuffer

/*
  split: 把一个流分离成两个
  select: 获取分离的其中一个流
 */
object SplitAndSelect {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val productData: DataStream[practice.Product] = CustomerSourceMySQL.mysqlConsumer(senv)

    val productTwoData: SplitStream[practice.Product] = productData.split(pro => {
      pro.pname match {
        case "裤子" => List("even")
        case "袜子" => List("odd")
        case _ => List("no")
      }
    })
    val oddData: DataStream[practice.Product] = productTwoData.select("odd")
    val allData: DataStream[practice.Product] = productTwoData.select("odd","even")
    oddData.print()
    senv.execute()
  }
}
